package com.crazywolf.kafka;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;


public class KafkaProducer extends Thread{

    private String topic;

    public KafkaProducer(String topic){
        super();
        this.topic = topic;
    }


    @Override
    public void run() {
        Producer producer = createProducer();
        int i=0;
        while(true){
            String  message= "message:"+i++;
            System.out.println(message);

            producer.send(new KeyedMessage<Integer, String>(topic, message));
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private Producer createProducer() {
        Properties properties = new Properties();
//        properties.put("zookeeper.connect", "192.168.4.185:2181,192.168.1.111:2181,192.168.1.112:2181");//????zk
        properties.put("zookeeper.connect", "192.168.4.186:2181");
        properties.put("serializer.class", StringEncoder.class.getName());
        properties.put("metadata.broker.list", "CrazyWolf:9092");
        return new Producer<Integer, String>(new ProducerConfig(properties));
    }




}